/*
 * Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 */
package io.airbyte.integrations.destination.bigquery

import com.fasterxml.jackson.databind.node.ObjectNode
import io.airbyte.cdk.load.util.Jsons
import io.airbyte.cdk.load.util.deserializeToNode
import io.airbyte.cdk.load.util.serializeToString
import io.airbyte.integrations.destination.bigquery.spec.BigqueryConfiguration
import io.airbyte.integrations.destination.bigquery.spec.BigqueryConfigurationFactory
import io.airbyte.integrations.destination.bigquery.spec.BigquerySpecification
import io.airbyte.integrations.destination.bigquery.spec.CdcDeletionMode
import java.io.IOException
import java.nio.file.Files
import java.nio.file.Path

object BigQueryDestinationTestUtils {
    const val RAW_DATASET_OVERRIDE = "overridden_raw_dataset"
    const val STANDARD_INSERT_CONFIG = "secrets/credentials-1s1t-standard.json"
    const val GCS_STAGING_CONFIG = "secrets/credentials-1s1t-gcs.json"

    val standardInsertRawOverrideConfig =
        createConfig(
            configFile = STANDARD_INSERT_CONFIG,
            rawDatasetId = RAW_DATASET_OVERRIDE,
        )
    val standardInsertConfig =
        createConfig(
            configFile = STANDARD_INSERT_CONFIG,
        )

    /**
     * Parse the config file and replace dataset with rawNamespace and stagingPath randomly
     * generated by the test.
     *
     * @param configFile Path to the config file
     */
    @Throws(IOException::class)
    fun createConfig(
        configFile: String,
        rawDatasetId: String? = null,
        disableTypingDeduping: Boolean? = null,
        cdcDeletionMode: CdcDeletionMode? = null,
    ): String {
        val tmpConfigAsString = Files.readString(Path.of(configFile))
        val config = Jsons.readTree(tmpConfigAsString) as ObjectNode
        config.put(BigQueryConsts.CONFIG_DATASET_ID, DEFAULT_NAMESPACE_PLACEHOLDER)
        rawDatasetId?.let { config.put(BigQueryConsts.RAW_DATA_DATASET, rawDatasetId) }
        disableTypingDeduping?.let {
            config.put(BigQueryConsts.DISABLE_TYPE_DEDUPE, disableTypingDeduping)
        }
        cdcDeletionMode?.let {
            config.put(BigQueryConsts.CDC_DELETION_MODE, cdcDeletionMode.cdcDeletionMode)
        }
        return config.serializeToString()
    }

    fun parseConfig(config: String): BigqueryConfiguration {
        val spec = Jsons.treeToValue(config.deserializeToNode(), BigquerySpecification::class.java)
        return BigqueryConfigurationFactory().make(spec)
    }
}
